package com.ada.spark.checkpoint

import org.apache.spark.{SparkConf, SparkContext}

/**
  * Spark中对于数据的保存除了持久化操作之外，还提供了一种检查点的机制，检查点（本质是通过将RDD写入Disk做检查点）是为了通过lineage做容错的辅助，lineage过长会造成容错成本过高，这样就不如在中间阶段做检查点容错，如果之后有节点出现问题而丢失分区，从做检查点的RDD开始重做Lineage，就会减少开销。检查点通过将数据写入到HDFS文件系统实现了RDD的检查点功能。
  * 为当前RDD设置检查点。该函数将会创建一个二进制的文件，并存储到checkpoint目录中，该目录是用SparkContext.setCheckpointDir()设置的。在checkpoint的过程中，该RDD的所有依赖于父RDD中的信息将全部被移除。对RDD进行checkpoint操作并不会马上被执行，必须执行Action操作才能触发。
  */
object RDDCheckPoint {
    def main(args: Array[String]): Unit = {
        //初始化配置信息及SparkContext
        val sparkConf: SparkConf = new SparkConf().setAppName("RDDCheckPoint").setMaster("local[*]")
        val sc = new SparkContext(sparkConf)

        sc.setCheckpointDir("hdfs://hadoop121:9000/checkpoint")
        val rdd = sc.parallelize(Array("ada"))
        val ch = rdd.map(_ + System.currentTimeMillis)
        println(ch.checkpoint)

        println(ch.collect().mkString("")); //ada1570960947563
        println(ch.collect().mkString("")); //ada1570960947798
        println(ch.collect().mkString("")); //ada1570960947798
    }
}
